from __future__ import with_statement

import os
import unittest

from tipfy import Request, RequestHandler, Response, Rule, Tipfy
from tipfy.app import local

import tipfy.auth
from tipfy.auth import (AdminRequiredMiddleware, LoginRequiredMiddleware,
    UserRequiredMiddleware, UserRequiredIfAuthenticatedMiddleware,
    admin_required, login_required, user_required,
    user_required_if_authenticated, check_password_hash, generate_password_hash,
    create_session_id, MultiAuthStore)
from tipfy.appengine.auth import AuthStore, MixedAuthStore
from tipfy.appengine.auth.model import User

import test_utils


class LoginHandler(RequestHandler):
    def get(self, **kwargs):
        return Response('login')


class LogoutHandler(RequestHandler):
    def get(self, **kwargs):
        return Response('logout')


class SignupHandler(RequestHandler):
    def get(self, **kwargs):
        return Response('signup')


class HomeHandler(RequestHandler):
    def get(self, **kwargs):
        return Response('home sweet home')


def get_app():
    app = Tipfy(rules=[
        Rule('/login', name='auth/login', handler=LoginHandler),
        Rule('/logout', name='auth/logout', handler=LogoutHandler),
        Rule('/signup', name='auth/signup', handler=SignupHandler),
    ])
    return app


class TestAuthStore(test_utils.BaseTestCase):
    def test_user_model(self):
        app = get_app()
        app.router.add(Rule('/', name='home', handler=HomeHandler))

        request = Request.from_values('/')
        local.current_handler = RequestHandler(app, request)

        store = AuthStore(local.current_handler)
        self.assertEqual(store.user_model, User)

    def test_login_url(self):
        app = get_app()
        app.router.add(Rule('/', name='home', handler=HomeHandler))

        with app.get_test_context() as request:
            request.app.router.match(request)

            store = AuthStore(request)
            self.assertEqual(store.login_url(), request.app.router.url_for(request, 'auth/login', dict(redirect='/')))

            tipfy.auth.DEV_APPSERVER_APPSERVER = False
            store.config['secure_urls'] = True
            self.assertEqual(store.login_url(), request.app.router.url_for(request, 'auth/login', dict(redirect='/', _scheme='https')))
            tipfy.auth.DEV_APPSERVER_APPSERVER = True

    def test_logout_url(self):
        app = get_app()
        app.router.add(Rule('/', name='home', handler=HomeHandler))

        with app.get_test_context() as request:
            request.app.router.match(request)
            store = AuthStore(request)
            self.assertEqual(store.logout_url(), request.app.router.url_for(request, 'auth/logout', dict(redirect='/')))

    def test_signup_url(self):
        app = get_app()
        app.router.add(Rule('/', name='home', handler=HomeHandler))

        with app.get_test_context() as request:
            request.app.router.match(request)
            store = AuthStore(request)
            self.assertEqual(store.signup_url(), request.app.router.url_for(request, 'auth/signup', dict(redirect='/')))


class TestMiddleware(test_utils.BaseTestCase):
    def tearDown(self):
        os.environ.pop('USER_EMAIL', None)
        os.environ.pop('USER_ID', None)
        os.environ.pop('USER_IS_ADMIN', None)
        test_utils.BaseTestCase.tearDown(self)

    def test_login_required_middleware_invalid(self):
        class MyHandler(HomeHandler):
            middleware = [LoginRequiredMiddleware()]

        app = get_app()
        app.router.add(Rule('/', name='home', handler=MyHandler))
        client = app.get_test_client()

        response = client.get('/')

        self.assertEqual(response.status_code, 302)
        self.assertEqual(response.headers['Location'], 'http://localhost/login?redirect=%2F')

    def test_login_required_decorator_invalid(self):
        class MyHandler(HomeHandler):
            @login_required
            def get(self, **kwargs):
                return Response('home sweet home')

        app = get_app()
        app.router.add(Rule('/', name='home', handler=MyHandler))
        client = app.get_test_client()

        response = client.get('/')

        self.assertEqual(response.status_code, 302)
        self.assertEqual(response.headers['Location'], 'http://localhost/login?redirect=%2F')

    def test_login_required_middleware(self):
        os.environ['USER_EMAIL'] = 'me@myself.com'
        os.environ['USER_ID'] = 'me'

        class MyHandler(HomeHandler):
            middleware = [LoginRequiredMiddleware()]

        app = get_app()
        app.router.add(Rule('/', name='home', handler=MyHandler))
        client = app.get_test_client()

        response = client.get('/')

        self.assertEqual(response.status_code, 200)
        self.assertEqual(response.data, 'home sweet home')

    def test_login_required_decorator(self):
        os.environ['USER_EMAIL'] = 'me@myself.com'
        os.environ['USER_ID'] = 'me'

        class MyHandler(HomeHandler):
            @login_required
            def get(self, **kwargs):
                return Response('home sweet home')

        app = get_app()
        app.router.add(Rule('/', name='home', handler=MyHandler))
        client = app.get_test_client()

        response = client.get('/')

        self.assertEqual(response.status_code, 200)
        self.assertEqual(response.data, 'home sweet home')

    def test_user_required_middleware_invalid(self):
        class MyHandler(HomeHandler):
            middleware = [UserRequiredMiddleware()]

        app = get_app()
        app.router.add(Rule('/', name='home', handler=MyHandler))
        client = app.get_test_client()

        response = client.get('/')

        self.assertEqual(response.status_code, 302)
        self.assertEqual(response.headers['Location'], 'http://localhost/login?redirect=%2F')

    def test_user_required_decorator_invalid(self):
        class MyHandler(HomeHandler):
            @user_required
            def get(self, **kwargs):
                return Response('home sweet home')

        app = get_app()
        app.router.add(Rule('/', name='home', handler=MyHandler))
        client = app.get_test_client()

        response = client.get('/')

        self.assertEqual(response.status_code, 302)
        self.assertEqual(response.headers['Location'], 'http://localhost/login?redirect=%2F')

    def test_user_required_middleware_logged_in(self):
        os.environ['USER_EMAIL'] = 'me@myself.com'
        os.environ['USER_ID'] = 'me'

        class MyHandler(HomeHandler):
            middleware = [UserRequiredMiddleware()]

        app = get_app()
        app.router.add(Rule('/', name='home', handler=MyHandler))
        client = app.get_test_client()

        response = client.get('/')

        self.assertEqual(response.status_code, 302)
        self.assertEqual(response.headers['Location'], 'http://localhost/signup?redirect=%2F')

    def test_user_required_decorator_logged_in(self):
        os.environ['USER_EMAIL'] = 'me@myself.com'
        os.environ['USER_ID'] = 'me'

        class MyHandler(HomeHandler):
            @user_required
            def get(self, **kwargs):
                return Response('home sweet home')

        app = get_app()
        app.router.add(Rule('/', name='home', handler=MyHandler))
        client = app.get_test_client()

        response = client.get('/')

        self.assertEqual(response.status_code, 302)
        self.assertEqual(response.headers['Location'], 'http://localhost/signup?redirect=%2F')

    def test_user_required_middleware_with_user(self):
        os.environ['USER_EMAIL'] = 'me@myself.com'
        os.environ['USER_ID'] = 'me'

        User.create('me', 'gae|me')

        class MyHandler(HomeHandler):
            middleware = [UserRequiredMiddleware()]

        app = get_app()
        app.router.add(Rule('/', name='home', handler=MyHandler))
        client = app.get_test_client()

        response = client.get('/')

        self.assertEqual(response.status_code, 200)
        self.assertEqual(response.data, 'home sweet home')

    def test_user_required_decorator_with_user(self):
        os.environ['USER_EMAIL'] = 'me@myself.com'
        os.environ['USER_ID'] = 'me'

        User.create('me', 'gae|me')

        class MyHandler(HomeHandler):
            @user_required
            def get(self, **kwargs):
                return Response('home sweet home')

        app = get_app()
        app.router.add(Rule('/', name='home', handler=MyHandler))
        client = app.get_test_client()

        response = client.get('/')

        self.assertEqual(response.status_code, 200)
        self.assertEqual(response.data, 'home sweet home')

    def test_user_required_if_authenticated_middleware(self):
        class MyHandler(HomeHandler):
            middleware = [UserRequiredIfAuthenticatedMiddleware()]

        app = get_app()
        app.router.add(Rule('/', name='home', handler=MyHandler))
        client = app.get_test_client()

        response = client.get('/')

        self.assertEqual(response.status_code, 200)
        self.assertEqual(response.data, 'home sweet home')

    def test_user_required_if_authenticate_decorator(self):
        class MyHandler(HomeHandler):
            @user_required_if_authenticated
            def get(self, **kwargs):
                return Response('home sweet home')

        app = get_app()
        app.router.add(Rule('/', name='home', handler=MyHandler))
        client = app.get_test_client()

        response = client.get('/')

        self.assertEqual(response.status_code, 200)
        self.assertEqual(response.data, 'home sweet home')

    def test_user_required_if_authenticated_middleware_logged_in(self):
        os.environ['USER_EMAIL'] = 'me@myself.com'
        os.environ['USER_ID'] = 'me'

        class MyHandler(HomeHandler):
            middleware = [UserRequiredIfAuthenticatedMiddleware()]

        app = get_app()
        app.router.add(Rule('/', name='home', handler=MyHandler))
        client = app.get_test_client()

        response = client.get('/')

        self.assertEqual(response.status_code, 302)
        self.assertEqual(response.headers['Location'], 'http://localhost/signup?redirect=%2F')

    def test_user_required_if_authenticate_decorator_logged_in(self):
        os.environ['USER_EMAIL'] = 'me@myself.com'
        os.environ['USER_ID'] = 'me'

        class MyHandler(HomeHandler):
            @user_required_if_authenticated
            def get(self, **kwargs):
                return Response('home sweet home')

        app = get_app()
        app.router.add(Rule('/', name='home', handler=MyHandler))
        client = app.get_test_client()

        response = client.get('/')

        self.assertEqual(response.status_code, 302)
        self.assertEqual(response.headers['Location'], 'http://localhost/signup?redirect=%2F')

    def test_user_required_if_authenticated_middleware_with_user(self):
        os.environ['USER_EMAIL'] = 'me@myself.com'
        os.environ['USER_ID'] = 'me'

        User.create('me', 'gae|me')

        class MyHandler(HomeHandler):
            middleware = [UserRequiredIfAuthenticatedMiddleware()]

        app = get_app()
        app.router.add(Rule('/', name='home', handler=MyHandler))
        client = app.get_test_client()

        response = client.get('/')

        self.assertEqual(response.status_code, 200)
        self.assertEqual(response.data, 'home sweet home')

    def test_user_required_if_authenticate_decorator_with_user(self):
        os.environ['USER_EMAIL'] = 'me@myself.com'
        os.environ['USER_ID'] = 'me'

        User.create('me', 'gae|me')

        class MyHandler(HomeHandler):
            @user_required_if_authenticated
            def get(self, **kwargs):
                return Response('home sweet home')

        app = get_app()
        app.router.add(Rule('/', name='home', handler=MyHandler))
        client = app.get_test_client()

        response = client.get('/')

        self.assertEqual(response.status_code, 200)
        self.assertEqual(response.data, 'home sweet home')

    def test_admin_required_middleware(self):
        class MyHandler(HomeHandler):
            middleware = [AdminRequiredMiddleware()]

        app = get_app()
        app.router.add(Rule('/', name='home', handler=MyHandler))
        client = app.get_test_client()

        response = client.get('/')

        self.assertEqual(response.status_code, 302)
        self.assertEqual(response.headers['Location'], 'http://localhost/login?redirect=%2F')

    def test_admin_required_decorator(self):
        class MyHandler(HomeHandler):
            @admin_required
            def get(self, **kwargs):
                return Response('home sweet home')

        app = get_app()
        app.router.add(Rule('/', name='home', handler=MyHandler))
        client = app.get_test_client()

        response = client.get('/')

        self.assertEqual(response.status_code, 302)
        self.assertEqual(response.headers['Location'], 'http://localhost/login?redirect=%2F')

    def test_admin_required_middleware_logged_in(self):
        os.environ['USER_EMAIL'] = 'me@myself.com'
        os.environ['USER_ID'] = 'me'

        class MyHandler(HomeHandler):
            middleware = [AdminRequiredMiddleware()]

        app = get_app()
        app.router.add(Rule('/', name='home', handler=MyHandler))
        client = app.get_test_client()

        response = client.get('/')

        self.assertEqual(response.status_code, 403)

    def test_admin_required_decorator_logged_in(self):
        os.environ['USER_EMAIL'] = 'me@myself.com'
        os.environ['USER_ID'] = 'me'

        class MyHandler(HomeHandler):
            @admin_required
            def get(self, **kwargs):
                return Response('home sweet home')

        app = get_app()
        app.router.add(Rule('/', name='home', handler=MyHandler))
        client = app.get_test_client()

        response = client.get('/')

        self.assertEqual(response.status_code, 403)

    def test_admin_required_middleware_with_user(self):
        os.environ['USER_EMAIL'] = 'me@myself.com'
        os.environ['USER_ID'] = 'me'

        User.create('me', 'gae|me')

        class MyHandler(HomeHandler):
            middleware = [AdminRequiredMiddleware()]

        app = get_app()
        app.router.add(Rule('/', name='home', handler=MyHandler))
        client = app.get_test_client()

        response = client.get('/')

        self.assertEqual(response.status_code, 403)

    def test_admin_required_decorator_withd_user(self):
        os.environ['USER_EMAIL'] = 'me@myself.com'
        os.environ['USER_ID'] = 'me'

        User.create('me', 'gae|me')

        class MyHandler(HomeHandler):
            @admin_required
            def get(self, **kwargs):
                return Response('home sweet home')

        app = get_app()
        app.router.add(Rule('/', name='home', handler=MyHandler))
        client = app.get_test_client()

        response = client.get('/')

        self.assertEqual(response.status_code, 403)

    def test_admin_required_middleware_with_admin(self):
        os.environ['USER_EMAIL'] = 'me@myself.com'
        os.environ['USER_ID'] = 'me'

        User.create('me', 'gae|me', is_admin=True)

        class MyHandler(HomeHandler):
            middleware = [AdminRequiredMiddleware()]

        app = get_app()
        app.router.add(Rule('/', name='home', handler=MyHandler))
        client = app.get_test_client()

        response = client.get('/')

        self.assertEqual(response.status_code, 200)
        self.assertEqual(response.data, 'home sweet home')

    def test_admin_required_decorator_with_admin(self):
        os.environ['USER_EMAIL'] = 'me@myself.com'
        os.environ['USER_ID'] = 'me'

        User.create('me', 'gae|me', is_admin=True)

        class MyHandler(HomeHandler):
            @admin_required
            def get(self, **kwargs):
                return Response('home sweet home')

        app = get_app()
        app.router.add(Rule('/', name='home', handler=MyHandler))
        client = app.get_test_client()

        response = client.get('/')

        self.assertEqual(response.status_code, 200)
        self.assertEqual(response.data, 'home sweet home')


class TestUserModel(test_utils.BaseTestCase):
    def test_create(self):
        user = User.create('my_username', 'my_id')
        self.assertEqual(isinstance(user, User), True)

        # Second one will fail to be created.
        user = User.create('my_username', 'my_id')
        self.assertEqual(user, None)

    def test_create_with_password_hash(self):
        user = User.create('my_username', 'my_id', password_hash='foo')

        self.assertEqual(isinstance(user, User), True)
        self.assertEqual(user.password, 'foo')

    def test_create_with_password(self):
        user = User.create('my_username', 'my_id', password='foo')

        self.assertEqual(isinstance(user, User), True)
        self.assertNotEqual(user.password, 'foo')
        self.assertEqual(len(user.password.split('$')), 3)

    def test_set_password(self):
        user = User.create('my_username', 'my_id', password='foo')
        self.assertEqual(isinstance(user, User), True)

        password = user.password

        user.set_password('bar')
        self.assertNotEqual(user.password, password)

        self.assertNotEqual(user.password, 'bar')
        self.assertEqual(len(user.password.split('$')), 3)

    def test_check_password(self):
        app = Tipfy()
        request = Request.from_values('/')
        local.current_handler = RequestHandler(app, request)

        user = User.create('my_username', 'my_id', password='foo')

        self.assertEqual(user.check_password('foo'), True)
        self.assertEqual(user.check_password('bar'), False)

    def test_check_session(self):
        app = Tipfy()
        request = Request.from_values('/')
        local.current_handler = RequestHandler(app, request)

        user = User.create('my_username', 'my_id', password='foo')

        session_id = user.session_id
        self.assertEqual(user.check_session(session_id), True)
        self.assertEqual(user.check_session('bar'), False)

    def test_get_by_username(self):
        user = User.create('my_username', 'my_id')
        user_1 = User.get_by_username('my_username')

        self.assertEqual(isinstance(user, User), True)
        self.assertEqual(isinstance(user_1, User), True)
        self.assertEqual(str(user.key()), str(user_1.key()))

    def test_get_by_auth_id(self):
        user = User.create('my_username', 'my_id')
        user_1 = User.get_by_auth_id('my_id')

        self.assertEqual(isinstance(user, User), True)
        self.assertEqual(isinstance(user_1, User), True)
        self.assertEqual(str(user.key()), str(user_1.key()))

    def test_unicode(self):
        user_1 = User(username='Calvin', auth_id='test', session_id='test')
        self.assertEqual(unicode(user_1), u'Calvin')

    def test_str(self):
        user_1 = User(username='Hobbes', auth_id='test', session_id='test')
        self.assertEqual(str(user_1), u'Hobbes')

    def test_eq(self):
        user_1 = User(key_name='test', username='Calvin', auth_id='test', session_id='test')
        user_2 = User(key_name='test', username='Calvin', auth_id='test', session_id='test')

        self.assertEqual(user_1, user_2)
        self.assertNotEqual(user_1, '')

    def test_ne(self):
        user_1 = User(key_name='test', username='Calvin', auth_id='test', session_id='test')
        user_2 = User(key_name='test_2', username='Calvin', auth_id='test', session_id='test')

        self.assertEqual((user_1 != user_2), True)

    def test_renew_session(self):
        app = Tipfy()
        request = Request.from_values('/')
        local.current_handler = RequestHandler(app, request)

        user = User.create('my_username', 'my_id')
        user.renew_session(max_age=86400)

    def test_renew_session_force(self):
        app = Tipfy()
        user = User.create('my_username', 'my_id')
        user.renew_session(force=True, max_age=86400)


class TestMiscelaneous(test_utils.BaseTestCase):
    def test_create_session_id(self):
        self.assertEqual(len(create_session_id()), 32)


class TestMultiAuthStore(test_utils.BaseTestCase):
    def get_app(self):
        app = Tipfy(config={'tipfy.sessions': {
            'secret_key': 'secret',
        }})
        return app

    def test_login_with_form_invalid(self):
        with self.get_app().get_test_context() as request:
            store = MultiAuthStore(request)
            res = store.login_with_form('foo', 'bar', remember=True)
            self.assertEqual(res, False)

    def test_login_with_form(self):
        user = User.create('foo', 'foo_id', password='bar')
        with self.get_app().get_test_context() as request:
            store = MultiAuthStore(request)
            res = store.login_with_form('foo', 'bar', remember=True)
            self.assertEqual(res, True)

            res = store.login_with_form('foo', 'bar', remember=False)
            self.assertEqual(res, True)

    def test_login_with_auth_id(self):
        with self.get_app().get_test_context() as request:
            store = MultiAuthStore(request)
            store.login_with_auth_id('foo_id', remember=False)

            user = User.create('foo', 'foo_id', password='bar')
            app = self.get_app()
            store.login_with_auth_id('foo_id', remember=True)

    def test_real_login(self):
        user = User.create('foo', 'foo_id', auth_remember=True)
        with self.get_app().get_test_context() as request:
            store = MultiAuthStore(request)
            store.login_with_auth_id('foo_id', remember=False)

            response = Response()
            request.session_store.save(response)

        with self.get_app().get_test_context('/', headers={
            'Cookie': '\n'.join(response.headers.getlist('Set-Cookie')),
        }) as request:
            store = MultiAuthStore(request)
            self.assertNotEqual(store.user, None)
            self.assertEqual(store.user.username, 'foo')
            self.assertEqual(store.user.auth_id, 'foo_id')

    def test_real_logout(self):
        user = User.create('foo', 'foo_id', auth_remember=True)
        with self.get_app().get_test_context() as request:
            store = MultiAuthStore(request)
            store.login_with_auth_id('foo_id', remember=False)

            response = Response()
            request.session_store.save(response)

        with self.get_app().get_test_context('/', headers={
            'Cookie': '\n'.join(response.headers.getlist('Set-Cookie')),
        }) as request:
            store = MultiAuthStore(request)
            self.assertNotEqual(store.user, None)
            self.assertEqual(store.user.username, 'foo')
            store.logout()

            response = Response()
            request.session_store.save(response)

        with self.get_app().get_test_context('/', headers={
            'Cookie': '\n'.join(response.headers.getlist('Set-Cookie')),
        }) as request:
            store = MultiAuthStore(request)
            self.assertEqual(store.session, None)

    def test_real_login_no_user(self):
        with self.get_app().get_test_context() as request:
            store = MultiAuthStore(request)
            user = store.create_user('foo', 'foo_id')
            store.login_with_auth_id('foo_id', remember=False)

            response = Response()
            request.session_store.save(response)
            user.delete()

        with self.get_app().get_test_context('/', headers={
            'Cookie': '\n'.join(response.headers.getlist('Set-Cookie')),
        }) as request:
            store = MultiAuthStore(request)
            self.assertEqual(store.session['id'], 'foo_id')
            self.assertEqual(store.user, None)

    def test_real_login_invalid(self):
        with self.get_app().get_test_context() as request:
            store = MultiAuthStore(request)
            self.assertEqual(store.user, None)
            self.assertEqual(store.session, None)


if __name__ == '__main__':
    test_utils.main()
